from influxdb_client import InfluxDBClient, Point, WritePrecision  # pip install influxdb_client
import coin2.service.order_detector.common as common
from influxdb_client.client.write_api import ASYNCHRONOUS
from threading import Lock
import json

_token = common.influxdb_token
_org = common.influxdb_org
_bucket = common.influxdb_bucket
_url = common.influxdb_url
_bucket_order_filter = common.influxdb_bucket_order_filter


class InfluxdbOutPut:
  def __init__(self, *, url=_url, token=_token, org=_org, bucket=_bucket):
    assert (isinstance(bucket, str))
    assert (isinstance(org, str))
    self._bucket = bucket
    self._org = org
    with InfluxDBClient(url=url, token=token, org=org) as client:
      self._write_api = client.write_api(write_options=ASYNCHRONOUS, error_callback=self.error)

  def error(conf, data: str, exception):
    print(f"Cannot write batch: {conf}, data: {data} due: {exception}")

  def write_data_spot(self, name, tag: dict, filed: dict, ts):
    point = Point(name)
    for k, v in tag.items():
      point.tag(k, v)
    for k, v in filed.items():
      point.field(k, v)

    point.time(ts, WritePrecision.NS)
    self._write_api.write(self._bucket, self._org, point)
    self._write_api.flush()

  def close(self):
    self._write_api.close()

  def __del__(self):
    self._write_api.close()


class InfluxdbExecuter:
  def __init__(self, *, url=_url, token=_token, org=_org, bucket=_bucket_order_filter):
    self._org = org
    self._bucket = bucket
    self._client = InfluxDBClient(url=url, token=token, org=org)

  def query(self, query):
    tables = self._client.query_api().query(query, org=self._org)
    js = tables.to_json()
    js = json.loads(js)
    return js


class InfluxdbTakeOut:
  def __init__(self):
    self._client = InfluxdbExecuter()

  def get_native_symbol_by_strategy_name(self, strategy_name: str, exchange: str):
    queue = 'from(bucket: "order_filter")'\
            '|> range(start: -2h)'\
            '|> filter(fn: (r) => r["_measurement"] == "order_filter" and r["exchange"] == "%s")'\
            '|> filter(fn: (r) => r["strategy_name"] == "%s")'\
            '|> duplicate(column: "native_symbol", as: "_value")'\
            '|> group()'\
            '|> unique()'\
            '|> keep(columns: ["native_symbol", "symbol"])' % (exchange, strategy_name)

    result = self._client.query(queue)
    symbols = []
    for itme in result:
      symbols.append(itme["native_symbol"])
    return symbols

  def get_machine_by_strategy_name(self, strategy_name: str):
    queue = 'from(bucket: "ogre_order")'\
            '|> range(start: -1h)'\
            '|> filter(fn: (r) => r["_measurement"] == "pnl")'\
            '|> filter(fn: (r) => r["_field"] == "pnl_balance_total")'\
            '|> filter(fn: (r) => r.strategy_name == "%s")'\
            '|> last()'\
            '|> keep(columns: ["strategy_name", "machine"])' % (strategy_name)
    resp = self._client.query(queue)
    machine = ""
    if len(resp) > 0:
      machine = resp[0]["machine"]

    return machine


if __name__ == "__main__":
  out = InfluxdbTakeOut()
  # machine = out.get_machine_by_strategy_name("rmm_prex_swap_prod_vet_0")
  # print(machine)
  native_symbols = out.get_native_symbol_by_strategy_name("hamm_joy_usdt", "Mexc")
  print(native_symbols)
